/*******************************************************************************
 * Includes
 ******************************************************************************/
#include "Wed_framework.h"
#include <open62541/server.h>
#include <server/ua_server_internal.h>
#include <open62541/plugin/log_stdout.h>

/*******************************************************************************
 * Marco operate
 ******************************************************************************/
#define ETH_PUBLISH_ADDRESS     "opc.eth://01-00-5E-00-00-01"
#define ETH_SUBSCRIBER_ADDRESS  "opc.eth://01-00-5E-00-00-02"
#define KEY_STREAM_NAME         "streamName"
#define KEY_STACK_IDX           "stackIdx"
#define DATA_SET_WRITER_ID       62541
#define PUBLISHER_ID_SUB         2234
#define WRITER_GROUP_ID_SUB      101

/*******************************************************************************
 * Static
 ******************************************************************************/
static struct Wed_thread *__g_p_pub_thread              = NULL;

static UA_NodeId            seqNumNodeId;
static UA_NodeId            cycleTriggerTimeNodeId;
static UA_NodeId            taskBeginTimeNodeId;
static UA_NodeId            taskEndTimeNodeId;

static UA_UInt32            sequenceNumber              = 0;
//static UA_UInt64            lastCycleTriggerTime        = 0;
//static UA_UInt64            lastTaskBeginTime           = 0;
//static UA_UInt64            lastTaskEndTime             = 0;
static UA_DataValue        *staticValueSeqNum           = NULL;
static UA_DataValue        *staticValueCycTrig          = NULL;
static UA_DataValue        *staticValueCycBegin         = NULL;
static UA_DataValue        *staticValueCycEnd           = NULL;

static UA_Boolean           running                     = true;
//static UA_Double            pubInterval                 = 0;

static UA_UInt32          *sequenceNumberSub            = NULL;
static UA_DataValue       *sequenceNumberSubDataValueRT = NULL;

/*******************************************************************************
 * Code
 ******************************************************************************/
/**
 * \brief 添加服务器节点
 */
static int addServerNodes(UA_Server *p_server) {
    int                   ret;
    UA_UInt64             initVal64                = 0;
    UA_UInt32             initVal32                = 0;
    UA_NodeId             folderId;
    UA_NodeId             folder_sub_Id;
    UA_ObjectAttributes   oAttr                    = UA_ObjectAttributes_default;
    UA_VariableAttributes publisherAttr            = UA_VariableAttributes_default;
    UA_ValueBackend       valueBackend;
    UA_VariableAttributes sequenceNumberSubscriber = UA_VariableAttributes_default;
    UA_UInt32             sequenceNumber           = 0;
    UA_NodeId             newNodeId;

    UA_NodeId_init(&folderId);

    oAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Publisher TSN");

    /* 创建一个目录对象 */
    ret = UA_Server_addObjectNode(p_server,
                                  UA_NODEID_NULL,
                                  UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
                                  UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES),
                                  UA_QUALIFIEDNAME(1, "Publisher TSN"),
                                  UA_NODEID_NUMERIC(0, UA_NS0ID_BASEOBJECTTYPE),
                                  oAttr,
                                  NULL,
                                 &folderId);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    /* 创建一个序列号节点 */
    UA_NodeId_init(&seqNumNodeId);
    seqNumNodeId = UA_NODEID_STRING(1, "sequence.number");
    publisherAttr.dataType = UA_TYPES[UA_TYPES_UINT32].typeId;
    UA_Variant_setScalar(&publisherAttr.value, &initVal32, &UA_TYPES[UA_TYPES_UINT32]);
    publisherAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Sequence Number");
    publisherAttr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;

    ret = UA_Server_addVariableNode(p_server,
                                    seqNumNodeId,
                                    folderId,
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
                                    UA_QUALIFIEDNAME(1, "Sequence Number"),
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),
                                    publisherAttr,
                                    NULL,
                                    NULL);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    /* 设置序列号节点后端数据 */
    valueBackend.backendType = UA_VALUEBACKENDTYPE_EXTERNAL;
    valueBackend.backend.external.value = &staticValueSeqNum;
    UA_Server_setVariableNode_valueBackend(p_server, seqNumNodeId, valueBackend);

    /* 创建一个循环触发时间节点 */
    UA_NodeId_init(&cycleTriggerTimeNodeId);
    cycleTriggerTimeNodeId = UA_NODEID_STRING(1, "cycle.trigger.time");
    publisherAttr = UA_VariableAttributes_default;
    publisherAttr.dataType = UA_TYPES[UA_TYPES_UINT64].typeId;
    UA_Variant_setScalar(&publisherAttr.value, &initVal64, &UA_TYPES[UA_TYPES_UINT64]);
    publisherAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Cycle Trigger Time");
    publisherAttr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;

    ret = UA_Server_addVariableNode(p_server,
                                    cycleTriggerTimeNodeId,
                                    folderId,
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
                                    UA_QUALIFIEDNAME(1, "Cycle Trigger Time"),
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),
                                    publisherAttr,
                                    NULL,
                                    NULL);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }
    /* 设置循环触发时间节点后端数据 */
    valueBackend.backend.external.value = &staticValueCycTrig;
    UA_Server_setVariableNode_valueBackend(p_server, cycleTriggerTimeNodeId, valueBackend);

    /* 创建一个任务起始时间节点 */
    UA_NodeId_init(&taskBeginTimeNodeId);
    taskBeginTimeNodeId = UA_NODEID_STRING(1, "task.begin.time");
    publisherAttr = UA_VariableAttributes_default;
    publisherAttr.dataType = UA_TYPES[UA_TYPES_UINT64].typeId;
    UA_Variant_setScalar(&publisherAttr.value, &initVal64, &UA_TYPES[UA_TYPES_UINT64]);
    publisherAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Task Begin Time");
    publisherAttr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;

    ret = UA_Server_addVariableNode(p_server,
                                    taskBeginTimeNodeId,
                                    folderId,
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
                                    UA_QUALIFIEDNAME(1, "Task Begin Time"),
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),
                                    publisherAttr,
                                    NULL,
                                    NULL);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }
    /* 设置任务起始时间节点后端数据 */
    valueBackend.backend.external.value = &staticValueCycBegin;
    UA_Server_setVariableNode_valueBackend(p_server, taskBeginTimeNodeId, valueBackend);

    /* 创建一个任务结束时间节点 */
    UA_NodeId_init(&taskEndTimeNodeId);
    taskEndTimeNodeId = UA_NODEID_STRING(1, "task.end.time");
    publisherAttr = UA_VariableAttributes_default;
    publisherAttr.dataType = UA_TYPES[UA_TYPES_UINT64].typeId;
    UA_Variant_setScalar(&publisherAttr.value, &initVal64, &UA_TYPES[UA_TYPES_UINT64]);
    publisherAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Task End Time");
    publisherAttr.accessLevel = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;

    ret = UA_Server_addVariableNode(p_server,
                                    taskEndTimeNodeId,
                                    folderId,
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
                                    UA_QUALIFIEDNAME(1, "Task End Time"),
                                    UA_NODEID_NUMERIC(0, UA_NS0ID_BASEDATAVARIABLETYPE),
                                    publisherAttr,
                                    NULL,
                                    NULL);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }
    /* 设置任务结束时间节点后端数据 */
    valueBackend.backend.external.value = &staticValueCycEnd;
    UA_Server_setVariableNode_valueBackend(p_server, taskEndTimeNodeId, valueBackend);

    UA_NodeId_init(&folderId);

    oAttr.displayName = UA_LOCALIZEDTEXT("en-US", "Subscriber TSN");

    /* 添加一个对象节点 */
    ret = UA_Server_addObjectNode(p_server,
                                  UA_NODEID_NULL,
                                  UA_NODEID_NUMERIC(0, UA_NS0ID_OBJECTSFOLDER),
                                  UA_NODEID_NUMERIC(0, UA_NS0ID_ORGANIZES),
                                  UA_QUALIFIEDNAME(1, "Subscriber TSN"),
                                  UA_NODEID_NULL,
                                  oAttr,
                                  NULL,
                                 &folder_sub_Id);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    /* 初始化变量属性 */
    sequenceNumberSubscriber.accessLevel           = UA_ACCESSLEVELMASK_READ | UA_ACCESSLEVELMASK_WRITE;
    UA_Variant_setScalar(&sequenceNumberSubscriber.value, &sequenceNumber, &UA_TYPES[UA_TYPES_UINT32]);
    sequenceNumberSubscriber.displayName           = UA_LOCALIZEDTEXT("en-US", "sequenceNumber Sub");
    sequenceNumberSubscriber.dataType              = UA_TYPES[UA_TYPES_UINT32].typeId;
    newNodeId                                      = UA_NODEID_NUMERIC(1, (UA_UInt32)30000);
    /* 添加服务器变量节点 */
    return UA_Server_addVariableNode(p_server,
                                     newNodeId,
                                     folder_sub_Id,
                                     UA_NODEID_NUMERIC(0, UA_NS0ID_HASCOMPONENT),
                                     UA_QUALIFIEDNAME(1, "sequenceNumber Sub"),
                                     UA_NODEID_NULL,
                                     sequenceNumberSubscriber,
                                     NULL,
                                     NULL);
}

/**
 * \brief 添加 PubSub 配置
 */
static int addPubSubConfiguration(UA_Server *p_server) {
    int                          ret;
    UA_NodeId                    connectionIdent;
    UA_NodeId                    publishedDataSetIdent;
    UA_NodeId                    writerGroupIdent;
    UA_PubSubConnectionConfig    connectionConfig;
    UA_NetworkAddressUrlDataType networkAddressUrl;
    // UA_KeyValuePair              connectionOptions[2];
    UA_PublishedDataSetConfig    publishedDataSetConfig;
    UA_DataSetFieldConfig        dataSetFieldCfg;
    UA_NodeId                    f4;
    //UA_NodeId                    f1, f2, f3, f4;
    UA_WriterGroupConfig         writerGroupConfig;
    UA_NodeId                    dataSetWriterIdent;
    UA_DataSetWriterConfig       dataSetWriterConfig;
    UA_AddPublishedDataSetResult pds_result;
    UA_DataSetFieldResult        dsf_result;

    memset(&connectionConfig, 0, sizeof(connectionConfig));

    connectionConfig.name                = UA_STRING("ETH Publish Connection");
    connectionConfig.transportProfileUri = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-eth-uadp");
    connectionConfig.enabled             = true;
    connectionConfig.publisherId.uint16  = PUBLISHER_ID_SUB;

    networkAddressUrl.networkInterface = UA_STRING("/dev/eth_tsn0");
    networkAddressUrl.url              = UA_STRING(ETH_PUBLISH_ADDRESS);

    UA_Variant_setScalar(&connectionConfig.address, &networkAddressUrl,
                         &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);
    connectionConfig.publisherId.uint32 = UA_UInt32_random();

    /* 添加 PubSub 连接 */
    ret = UA_Server_addPubSubConnection(p_server, &connectionConfig, &connectionIdent);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    memset(&publishedDataSetConfig, 0, sizeof(UA_PublishedDataSetConfig));
    publishedDataSetConfig.publishedDataSetType = UA_PUBSUB_DATASET_PUBLISHEDITEMS;
    publishedDataSetConfig.name                 = UA_STRING("Demo PDS");
    /* 添加发布数据集 */
    pds_result = UA_Server_addPublishedDataSet(p_server,
                                              &publishedDataSetConfig,
                                              &publishedDataSetIdent);
    if (pds_result.addResult != UA_STATUSCODE_GOOD) {
        return ret;
    }

    /* 初始化发布数据集的字段 */
    memset(&dataSetFieldCfg, 0, sizeof(UA_DataSetFieldConfig));
    dataSetFieldCfg.dataSetFieldType = UA_PUBSUB_DATASETFIELD_VARIABLE;
    dataSetFieldCfg.field.variable.fieldNameAlias = UA_STRING ("Sequence Number");
    dataSetFieldCfg.field.variable.promotedField = UA_FALSE;
    dataSetFieldCfg.field.variable.publishParameters.publishedVariable = seqNumNodeId;
    dataSetFieldCfg.field.variable.publishParameters.attributeId = UA_ATTRIBUTEID_VALUE;

    dataSetFieldCfg.field.variable.rtValueSource.rtFieldSourceEnabled = UA_TRUE;
    staticValueSeqNum = UA_DataValue_new();
    UA_Variant_setScalar(&staticValueSeqNum->value, &sequenceNumber, &UA_TYPES[UA_TYPES_UINT32]);
    staticValueSeqNum->value.storageType = UA_VARIANT_DATA_NODELETE;
    dataSetFieldCfg.field.variable.rtValueSource.staticValueSource = &staticValueSeqNum;

    /* 添加发布数据集字段 */
    dsf_result = UA_Server_addDataSetField(p_server, publishedDataSetIdent, &dataSetFieldCfg, &f4);
    if (dsf_result.result != UA_STATUSCODE_GOOD) {
        return ret;
    }

//    memset(&dataSetFieldCfg, 0, sizeof(UA_DataSetFieldConfig));
//    dataSetFieldCfg.dataSetFieldType = UA_PUBSUB_DATASETFIELD_VARIABLE;
//    dataSetFieldCfg.field.variable.fieldNameAlias = UA_STRING ("Cycle Trigger Time");
//    dataSetFieldCfg.field.variable.promotedField = UA_FALSE;
//    dataSetFieldCfg.field.variable.publishParameters.publishedVariable = cycleTriggerTimeNodeId;
//    dataSetFieldCfg.field.variable.publishParameters.attributeId = UA_ATTRIBUTEID_VALUE;
//
//    dataSetFieldCfg.field.variable.rtValueSource.rtFieldSourceEnabled = UA_TRUE;
//    staticValueCycTrig = UA_DataValue_new();
//    UA_Variant_setScalar(&staticValueCycTrig->value, &lastCycleTriggerTime, &UA_TYPES[UA_TYPES_UINT64]);
//    staticValueCycTrig->value.storageType = UA_VARIANT_DATA_NODELETE;
//    dataSetFieldCfg.field.variable.rtValueSource.staticValueSource = &staticValueCycTrig;
//
//    /* 添加发布数据集字段 */
//    dsf_result = UA_Server_addDataSetField(p_server, publishedDataSetIdent, &dataSetFieldCfg, &f3);
//    if (dsf_result.result != UA_STATUSCODE_GOOD) {
//        return ret;
//    }
//
//    memset(&dataSetFieldCfg, 0, sizeof(UA_DataSetFieldConfig));
//    dataSetFieldCfg.dataSetFieldType = UA_PUBSUB_DATASETFIELD_VARIABLE;
//    dataSetFieldCfg.field.variable.fieldNameAlias = UA_STRING ("Task Begin Time");
//    dataSetFieldCfg.field.variable.promotedField = UA_FALSE;
//    dataSetFieldCfg.field.variable.publishParameters.publishedVariable = taskBeginTimeNodeId;
//    dataSetFieldCfg.field.variable.publishParameters.attributeId = UA_ATTRIBUTEID_VALUE;
//
//    dataSetFieldCfg.field.variable.rtValueSource.rtFieldSourceEnabled = UA_TRUE;
//    staticValueCycBegin = UA_DataValue_new();
//    UA_Variant_setScalar(&staticValueCycBegin->value, &lastTaskBeginTime, &UA_TYPES[UA_TYPES_UINT64]);
//    staticValueCycBegin->value.storageType = UA_VARIANT_DATA_NODELETE;
//    dataSetFieldCfg.field.variable.rtValueSource.staticValueSource = &staticValueCycBegin;
//
//    /* 添加发布数据集字段 */
//    dsf_result = UA_Server_addDataSetField(p_server, publishedDataSetIdent, &dataSetFieldCfg, &f2);
//    if (dsf_result.result != UA_STATUSCODE_GOOD) {
//        return ret;
//    }
//
//    memset(&dataSetFieldCfg, 0, sizeof(UA_DataSetFieldConfig));
//    dataSetFieldCfg.dataSetFieldType = UA_PUBSUB_DATASETFIELD_VARIABLE;
//    dataSetFieldCfg.field.variable.fieldNameAlias = UA_STRING ("Task End Time");
//    dataSetFieldCfg.field.variable.promotedField = UA_FALSE;
//    dataSetFieldCfg.field.variable.publishParameters.publishedVariable = taskEndTimeNodeId;
//    dataSetFieldCfg.field.variable.publishParameters.attributeId = UA_ATTRIBUTEID_VALUE;
//
//    dataSetFieldCfg.field.variable.rtValueSource.rtFieldSourceEnabled = UA_TRUE;
//    staticValueCycEnd = UA_DataValue_new();
//    UA_Variant_setScalar(&staticValueCycEnd->value, &lastTaskEndTime, &UA_TYPES[UA_TYPES_UINT64]);
//    staticValueCycEnd->value.storageType = UA_VARIANT_DATA_NODELETE;
//    dataSetFieldCfg.field.variable.rtValueSource.staticValueSource = &staticValueCycEnd;
//
//    /* 添加发布数据集字段 */
//    dsf_result = UA_Server_addDataSetField(p_server, publishedDataSetIdent, &dataSetFieldCfg, &f1);
//    if (dsf_result.result != UA_STATUSCODE_GOOD) {
//        return ret;
//    }

    memset(&writerGroupConfig, 0, sizeof(UA_WriterGroupConfig));
    writerGroupConfig.name = UA_STRING("Demo WriterGroup");
    writerGroupConfig.publishingInterval = 1000;
    writerGroupConfig.enabled            = UA_FALSE;
    writerGroupConfig.writerGroupId      = WRITER_GROUP_ID_SUB;
    writerGroupConfig.encodingMimeType   = UA_PUBSUB_ENCODING_UADP;
    writerGroupConfig.rtLevel            = UA_PUBSUB_RT_DETERMINISTIC;
    //writerGroupConfig.pubsubManagerCallback.addCustomCallback    = addApplicationCallback;
    //writerGroupConfig.pubsubManagerCallback.changeCustomCallback = changeApplicationCallbackInterval;
    //writerGroupConfig.pubsubManagerCallback.removeCustomCallback = removeApplicationPubSubCallback;

    /* 添加 WriterGroup */
    ret = UA_Server_addWriterGroup(p_server, connectionIdent,
                                  &writerGroupConfig, &writerGroupIdent);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    memset(&dataSetWriterConfig, 0, sizeof(UA_DataSetWriterConfig));
    dataSetWriterConfig.name = UA_STRING("Demo DataSetWriter");
    dataSetWriterConfig.dataSetWriterId = DATA_SET_WRITER_ID;
    dataSetWriterConfig.keyFrameCount = 10;

    /* 添加 DataSetWriter */
    ret = UA_Server_addDataSetWriter(p_server, writerGroupIdent, publishedDataSetIdent,
                                     &dataSetWriterConfig, &dataSetWriterIdent);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }
    /* 冻结 WriterGroup 配置 */
    UA_Server_freezeWriterGroupConfiguration(p_server, writerGroupIdent);
    /* 设置 WriterGroup 为 operational 状态 */
    UA_Server_setWriterGroupOperational(p_server, writerGroupIdent);

    return 0;
}

/**
 * \brief 外部数据写回调函数
 */
static UA_StatusCode externalDataWriteCallback(UA_Server *server, const UA_NodeId *sessionId,
                                               void *sessionContext, const UA_NodeId *nodeId,
                                               void *nodeContext, const UA_NumericRange *range,
                                               const UA_DataValue *data){
    int            ret;
    uint32_t       old_value;
    const UA_Node *p_node = UA_NODESTORE_GET(server, nodeId);

    old_value = *sequenceNumberSub;
    /* 清楚原来的值 */
    UA_DataValue_clear(*p_node->variableNode.valueBackend.backend.external.value);

    /* 复制新的值 */
    ret = UA_DataValue_copy(data, *p_node->variableNode.valueBackend.backend.external.value);
    if (ret == UA_STATUSCODE_GOOD) {
        WED_INFO_PRINT("OPC UA server subscriber sequence number：%d to %d", old_value, *sequenceNumberSub);
        sequenceNumber++;
    }
    return UA_STATUSCODE_GOOD;
}

static UA_StatusCode externalDataReadNotificationCallback(UA_Server *server, const UA_NodeId *sessionId,
                                                          void *sessionContext, const UA_NodeId *nodeid,
                                                          void *nodeContext, const UA_NumericRange *range){
    //allow read without any preparation
    return UA_STATUSCODE_GOOD;
}


/**
 * \brief 添加 PubSub 订阅配置
 */
static int addPubSubSubscriberConfiguration(UA_Server *p_server){
    UA_StatusCode                        ret                  = UA_STATUSCODE_GOOD;
    UA_NetworkAddressUrlDataType         networkAddressUrlsubscribe;
    UA_PubSubConnectionConfig            connectionConfig;
    UA_NodeId                            connectionIdentSubscriber;
    UA_ReaderGroupConfig                 readerGroupConfig;
    UA_NodeId                            readerGroupIdentifier;
    UA_DataSetReaderConfig               readerConfig;
    UA_UInt16                            publisherIdentifier  = PUBLISHER_ID_SUB;
    UA_UadpDataSetReaderMessageDataType *dataSetReaderMessage = NULL;
    UA_DataSetMetaDataType              *pMetaData            = NULL;
    UA_NodeId                            readerIdentifier;
    UA_ValueBackend                      valueBackend         = {0};

    memset(&connectionConfig, 0, sizeof(connectionConfig));
    connectionConfig.name                = UA_STRING("ETH Subscriber Connection");
    connectionConfig.enabled             = UA_TRUE;
    connectionConfig.transportProfileUri = UA_STRING("http://opcfoundation.org/UA-Profile/Transport/pubsub-eth-uadp");

    networkAddressUrlsubscribe.url              = UA_STRING(ETH_SUBSCRIBER_ADDRESS);
    networkAddressUrlsubscribe.networkInterface = UA_STRING("/dev/eth_tsn2");

    UA_Variant_setScalar(&connectionConfig.address, &networkAddressUrlsubscribe, &UA_TYPES[UA_TYPES_NETWORKADDRESSURLDATATYPE]);

    connectionConfig.publisherIdType    = UA_PUBLISHERIDTYPE_UINT32;
    /* 生成一个 32 位随机数 */
    connectionConfig.publisherId.uint32 = UA_UInt32_random();

    /* 添加 PubSub 连接 */
    ret |= UA_Server_addPubSubConnection(p_server, &connectionConfig, &connectionIdentSubscriber);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    memset (&readerGroupConfig, 0, sizeof(UA_ReaderGroupConfig));
    readerGroupConfig.name    = UA_STRING("Demo ReaderGroup");
    readerGroupConfig.rtLevel = UA_PUBSUB_RT_DETERMINISTIC;

    ret = UA_Server_addReaderGroup(p_server, connectionIdentSubscriber, &readerGroupConfig,
                                   &readerGroupIdentifier);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    memset (&readerConfig, 0, sizeof(UA_DataSetReaderConfig));
    readerConfig.name                 = UA_STRING("Demo DataSetReader");    /* 数据集订阅信息名称 */
    readerConfig.publisherId.type     = &UA_TYPES[UA_TYPES_UINT16];         /* 发布者 ID 类型 */
    readerConfig.publisherId.data     = &publisherIdentifier;               /* 发布者 ID  */
    readerConfig.writerGroupId        = WRITER_GROUP_ID_SUB;                /* 发布者 writerGroup ID */
    readerConfig.dataSetWriterId      = DATA_SET_WRITER_ID;                 /* 发布者 writerGroup 数据集 ID */

    readerConfig.messageSettings.encoding = UA_EXTENSIONOBJECT_DECODED; /* 无编码数据 */
    readerConfig.messageSettings.content.decoded.type = &UA_TYPES[UA_TYPES_UADPDATASETREADERMESSAGEDATATYPE];  /* 数据类型为 OPC UA 定义的二进制网络消息编码格式 */

    /* 分配一个新的 UADP DataSetReader 消息数据类型结构体 */
    dataSetReaderMessage = UA_UadpDataSetReaderMessageDataType_new();
    /* 设置 UADP DataSetReader 消息头部信息所包含的信息
     * 包含发布者 ID，GroupHeader，WriterGroup ID 和 PayloadHeader */
    dataSetReaderMessage->networkMessageContentMask = (UA_UadpNetworkMessageContentMask)(UA_UADPNETWORKMESSAGECONTENTMASK_PUBLISHERID |
                                                      (UA_UadpNetworkMessageContentMask)UA_UADPNETWORKMESSAGECONTENTMASK_GROUPHEADER |
                                                      (UA_UadpNetworkMessageContentMask)UA_UADPNETWORKMESSAGECONTENTMASK_WRITERGROUPID |
                                                      (UA_UadpNetworkMessageContentMask)UA_UADPNETWORKMESSAGECONTENTMASK_PAYLOADHEADER);
    readerConfig.messageSettings.content.decoded.data = dataSetReaderMessage;

    /* 初始化元数据配置 */
    pMetaData = &readerConfig.dataSetMetaData;
    UA_DataSetMetaDataType_init(pMetaData);

    /* 初始化字段元数据 */
    pMetaData->fieldsSize = 1;
    pMetaData->fields     = (UA_FieldMetaData*)UA_Array_new(pMetaData->fieldsSize, &UA_TYPES[UA_TYPES_FIELDMETADATA]);
    /* 第一个字段数据类型为 UINT32 类型 */
    UA_FieldMetaData_init(&pMetaData->fields[0]);
    UA_NodeId_copy(&UA_TYPES[UA_TYPES_UINT32].typeId, &pMetaData->fields[0].dataType);
    pMetaData->fields[0].builtInType = UA_NS0ID_UINT32;
    pMetaData->fields[0].valueRank   = -1;                /* 标量 */

    UA_FieldTargetVariable *targetVars = (UA_FieldTargetVariable *)UA_calloc(1, sizeof(UA_FieldTargetVariable));
    if (!targetVars) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "FieldTargetVariable - Bad out of memory");
        return -WED_ENOMEM;
    }

    /* 分配一个 UINT32 类型数据 */
    sequenceNumberSub = UA_UInt32_new();
    if (!sequenceNumberSub) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "sequenceNumberSub - Bad out of memory");
        UA_free(targetVars);
        return -WED_ENOMEM;;
    }

    /* 分配一个数据值结构体 */
    *sequenceNumberSub = 0;
    sequenceNumberSubDataValueRT = UA_DataValue_new();
    if (!sequenceNumberSubDataValueRT) {
        UA_LOG_ERROR(UA_Log_Stdout, UA_LOGCATEGORY_SERVER, "runningsubDatavalue - Bad out of memory");
        UA_free(targetVars);
        return -WED_ENOMEM;;
    }
    /* 设置数据值结构体为标量数据 */
    UA_Variant_setScalar(&sequenceNumberSubDataValueRT->value, sequenceNumberSub, &UA_TYPES[UA_TYPES_UINT32]);
    sequenceNumberSubDataValueRT->hasValue = UA_TRUE;

    /* 将上述创建节点的值后端设置为 “外部值源” */
    valueBackend.backendType = UA_VALUEBACKENDTYPE_EXTERNAL;          /* 引用外部的内存 */
    valueBackend.backend.external.value = &sequenceNumberSubDataValueRT;
    valueBackend.backend.external.callback.userWrite        = externalDataWriteCallback;
    valueBackend.backend.external.callback.notificationRead = externalDataReadNotificationCallback;
    UA_Server_setVariableNode_valueBackend(p_server, UA_NODEID_NUMERIC(1, (UA_UInt32)30000), valueBackend);

    UA_FieldTargetDataType_init(&targetVars[0].targetVariable);
    targetVars[0].targetVariable.attributeId  = UA_ATTRIBUTEID_VALUE;
    targetVars[0].targetVariable.targetNodeId = UA_NODEID_NUMERIC(1, (UA_UInt32)30000);

    /* 设置订阅数据的目标变量 */
    readerConfig.subscribedDataSetType = UA_PUBSUB_SDS_TARGET;
    readerConfig.subscribedDataSet.subscribedDataSetTarget.targetVariables = targetVars;
    readerConfig.subscribedDataSet.subscribedDataSetTarget.targetVariablesSize = 1;

    /* 初始化 DataSetReader 的元数据配置 */
    ret = UA_Server_addDataSetReader(p_server, readerGroupIdentifier, &readerConfig,
                                     &readerIdentifier);
    if (ret != UA_STATUSCODE_GOOD) {
        return ret;
    }

    UA_free(readerConfig.subscribedDataSet.subscribedDataSetTarget.targetVariables);
    UA_free(readerConfig.dataSetMetaData.fields);
    UA_UadpDataSetReaderMessageDataType_delete(dataSetReaderMessage);

    /* 冻结 ReaderGroup 配置 */
    UA_Server_freezeReaderGroupConfiguration(p_server, readerGroupIdentifier);
    /* 设置 ReaderGroup 为正在操作中状态 */
    UA_Server_setReaderGroupOperational(p_server, readerGroupIdentifier);

    return 0;
}

/**
 * \brief 应用任务函数
 */
static void __opcua_pub_task(void *p_arg){
    int ret;

    /* 创建服务器 */
    UA_Server *p_server = UA_Server_new();
    if (p_server == NULL) {
        WED_ERR_PRINT("OPC UA server create failed");
        goto __poll;
    }
    WED_INFO_PRINT("OPC UA server create successful");

    /* 添加服务器节点 */
    ret = addServerNodes(p_server);
    if (ret != UA_STATUSCODE_GOOD) {
        WED_ERR_PRINT("OPC UA server nodes add failed(%d)", ret);
        goto __poll;
    }
    WED_INFO_PRINT("OPC UA server nodes add successful");

    /* 添加 PubSub 订阅配置 */
    ret = addPubSubSubscriberConfiguration(p_server);
    if (ret != UA_STATUSCODE_GOOD) {
        WED_ERR_PRINT("OPC UA server PubSub connection subscriber add failed(%d)", ret);
        goto __poll;
    }
    WED_INFO_PRINT("OPC UA server PubSub connection subscriber add successful");

    /* 添加 PubSub 配置 */
    ret = addPubSubConfiguration(p_server);
    if (ret != UA_STATUSCODE_GOOD) {
        WED_ERR_PRINT("OPC UA server PubSub configuration add failed(%d)", ret);
        goto __poll;
    }
    WED_INFO_PRINT("OPC UA server PubSub configuration add successful");

    /* 运行服务器 */
    (void) UA_Server_run(p_server, &running);
    WED_ERR_PRINT("OPC UA server run failed");
__poll:
    while(1) {
        Wed_kernel_mdelay(1000);
    }
}

/**
 * \brief OPC UA PUBSUB TSN 回环
 *
 * @return 成功返回 0
 */
int opcua_pubsub_tsn_loopback(void){
    /* 创建线程 */
    __g_p_pub_thread = Wed_thread_create(__opcua_pub_task,
                                           NULL,
                                          "opcua_pub",
                                           10240,
                                           0);
    if (__g_p_pub_thread == NULL) {
        WED_ERR_PRINT("OPCUA pub thread create failed");
        return -WED_ECREAT;
    }
    return 0;
}
